package com.gblfy.kafka.controller;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

@RestController
public class KafkaSyncController {
    private final static Logger log = LoggerFactory.getLogger(KafkaSyncController.class);

    @Autowired
    private KafkaTemplate<Integer, String> kafkaTemplate;


    @RequestMapping("/send/sync/{message}")
    public String send(@PathVariable String message) {

        //同步发送消息
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("topic-springboot-01", 0, 0, message);
        try {
            SendResult<Integer, String> sendResult = future.get();
            RecordMetadata metadata = sendResult.getRecordMetadata();
            log.info("发送的主题:{} ,发送的分区:{} ,发送的偏移量:{} ",
                    metadata.topic(), metadata.partition(), metadata.offset());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return "success";
    }

}
